package edu.vt.ece.rt.quorum.messsages.types;

import java.util.HashSet;
import java.util.List;
import java.util.PriorityQueue;
import java.util.Set;

import org.jgroups.Address;
import org.jgroups.Message;


import edu.vt.ece.rt.dstm.Context;
import edu.vt.ece.rt.dstm.ContextDelegator;
import edu.vt.ece.rt.dstm.QObject;
import edu.vt.ece.rt.dstm.TMProxy;
import edu.vt.ece.rt.quorum.ClustersManager;
import edu.vt.ece.rt.quorum.Resource;
import edu.vt.ece.rt.quorum.messsages.QMessage;
import edu.vt.ece.rt.quorum.messsages.QMessage.MessageType;
import edu.vt.ece.rt.quorum.util.Names;
import edu.vt.ece.rt.util.Logger.Logger;

/*
 * Commit Request Message to commit DSTM transaction
 * @author Sudhanshu
 */
public class CommitRequest extends QMessage implements Runnable {
	/*
	 * Comparing at local level can be incorrect as version update can be
	 * for this node leading to needless aborts
	 */
	Context cxt; 			//Context containing the Read-Write Set of objects
	int id;					//Requester node of commit request
	
	public CommitRequest(Context c, int i) {
		super(MessageType.CommitRequest);
		cxt = c;
		id = i;	
	}
	
	@Override
	public void run() {
		Logger.debug("Commit request from "+id+" for tnx "+cxt.getContextId());
		//Remove the Transaction from all PR and PW list
		Resource.rmFrLists(cxt);
		
		Long txn = cxt.getContextId();
		if(ContextDelegator.getContext(txn).isAborted()){
			try {
				Logger.debug("tnx " + txn + " is already aborted, commit response abort");
				ClustersManager.getCluster(id + Names.writeQuorum).send(new Message(getSrcAddr(), new CommitResponse(null,txn, false) ));
				//send(new Message CommitResponse(src_addr, null, txn, false));
			} catch (Exception e) {
				e.printStackTrace();
			}
			return;
		}
		

		//Create conflicting list
		Set<Context> conflictSet = new HashSet<Context>();
		
		//Add yourself in conflictSet
		conflictSet.add(cxt);
		HashSet<String> lockedObjects = new HashSet<String>();
		
		for(QObject wobj: cxt.getWriteSet()){
			//Check local object status and version comparison
			if( Resource.getQObject(wobj.getId()).isProtected(txn) || isStaleVersion(wobj)){
				//True, means abort this transaction
				ContextDelegator.getContext(txn).setAborted();
				try {
					Logger.debug("Object " + wobj.getId()+ " is already locked by "+ Resource.getQObject(wobj.getId()).getOwner()+" or stale, aborting tnx " +txn);
					ClustersManager.getCluster(id + Names.writeQuorum).send(new Message(getSrcAddr(), new CommitResponse(null, txn, false) ));
					return;
				} catch (Exception e) {
					e.printStackTrace();
				}finally{
					//Unlock all previous taken objects
					unlockObjects(lockedObjects, txn);					
				}				
			}
			
			//Add PR and PW to conflicting list
			List<Context> l = Resource.getPRList(wobj.getId());			
			if(l!=null)
				conflictSet.addAll(l);
			l = Resource.getPWList(wobj.getId());
			if(l!=null)
				conflictSet.addAll(l);
							
			//Check if this transaction is highest priority transaction
			PriorityQueue<Context> prC = new PriorityQueue<Context>(conflictSet);
			Context hCxt = prC.peek();
			if(hCxt.getContextId().equals(txn)){		//Means still highest priority
				if(Resource.getQObject(wobj.getId()).lockObject(txn)){			//Now try lock 
					lockedObjects.add(wobj.getId());	//If you get it add to locked object list
					Logger.debug(txn + " got lock on "+ wobj.getId());
					continue;
				}
				Logger.debug("Object " + wobj.getId()+ " can not be locked, locked by "+wobj.getOwner()+" aborting tnx " +txn);
			}
			
			Logger.debug(txn + " is not highest Priority, highest is "+hCxt.getContextId() + " getting aborted");
			//If not continued must be aborted
			ContextDelegator.getContext(txn).setAborted();
			try {
				ClustersManager.getCluster(id + Names.writeQuorum).send(new Message(getSrcAddr(), new CommitResponse(null, txn, false) ));
				return;
			} catch (Exception e) {
				e.printStackTrace();
			}finally{
				//Unlock all previous taken objects
				unlockObjects(lockedObjects, txn);					
			}	
			
		}
		
		//Now check for the read Set
		for(QObject wobj: cxt.getReadSet()){
			if(Resource.getQObject(wobj.getId()).isProtected(txn) || isStaleVersion(wobj)){
				//True, means abort this transaction
				ContextDelegator.getContext(txn).setAborted();
				try {
					ClustersManager.getCluster(id + Names.writeQuorum).send(new Message(getSrcAddr(), new CommitResponse(null, txn, false) ));
					return;
				} catch (Exception e) {
					e.printStackTrace();
				}finally{
					//Unlock all previous taken objects
					unlockObjects(lockedObjects, txn);				
				}				
			}
			
			//Add PW to conflicting list
			List<Context> l = Resource.getPWList(wobj.getId());
			if(l!=null)
				conflictSet.addAll(l);
							
			//Check if this transaction is highest priority transaction
			PriorityQueue<Context> prC = new PriorityQueue<Context>(conflictSet);
			if(prC.peek().getContextId().equals(txn))		//Means still highest priority
				continue;
			
			//If not continued must be aborted
			ContextDelegator.getContext(txn).setAborted();
			try {
				Logger.debug("Read Validation Failed for" + wobj+" aborting tnx "+txn);
				ClustersManager.getCluster(id + Names.writeQuorum).send(new Message(getSrcAddr(), new CommitResponse(null, txn, false) ));
				return;
			} catch (Exception e) {
				e.printStackTrace();
			}finally{
				//Unlock all previous taken objects
				unlockObjects(lockedObjects, txn);					
			}	
		}
		
		//Check if Context got aborted
		if(ContextDelegator.getContext(txn).isAborted()){
			try {
				ClustersManager.getCluster(id + Names.writeQuorum).send(new Message(getSrcAddr(), new CommitResponse(null, txn, false) ));
				return;
			} catch (Exception e) {
				e.printStackTrace();
			}finally{
				//Unlock all previous taken objects
				unlockObjects(lockedObjects, txn);	
			}
		}
		
		//If passed the read and write both validation, it should be committed
		//remove yourself from conflicting list
		conflictSet.remove(cxt);
		//Abort all the other transaction in conflict list
		for(Context c: conflictSet){
			c.setAborted();
		}
		
		
		CommitResponse ca = new CommitResponse(conflictSet, txn, true);
		try {
			Logger.debug("Successfully verified tnx "+txn);
			ClustersManager.getCluster(id + Names.writeQuorum).send(new Message(getSrcAddr(),ca));
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * @param lockedObjects
	 */
	private void unlockObjects(HashSet<String> lockedObjects, Long txn) {
		for(String id: lockedObjects){
			Resource.getQObject(id).unlockObject(txn);		
			Logger.debug(txn + " left the lock on "+id);
		}
	}

	/**Stale Version 
	 * @param wobj
	 * @param cxt
	 * @return
	 */
	private boolean isStaleVersion(QObject wobj) {
		boolean b =  wobj.getVersion() < Resource.getQObject(wobj.getId()).getVersion();
		Logger.debug("Object is stale="+ b);
		return b;
	}	
}
